package net.wlm.others;

import com.alibaba.fastjson.JSON;
import net.wlm.utils.JdbcUtil;
import net.wlm.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class vehicleWindowBatteryHealthJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle_battery_data");
        BatteryHealth(kafkaStream);
        env.execute("vehicleWindowBatteryHealth");
    }

    private static void BatteryHealth(DataStream<String> kafkaStream) {
        SingleOutputStreamOperator<BatteryHealthData> mapStream = kafkaStream.map(o -> JSON.parseObject(o, BatteryHealthData.class));
        SingleOutputStreamOperator<BatteryHealthData> batteryHealthDataSingleOutputStreamOperator = mapStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<BatteryHealthData>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withTimestampAssigner(
                        new SerializableTimestampAssigner<BatteryHealthData>() {
                            @Override
                            public long extractTimestamp(BatteryHealthData element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }
                )
        );
        WindowedStream<BatteryHealthData, String, TimeWindow> window = batteryHealthDataSingleOutputStreamOperator.keyBy(o -> "all").window(TumblingEventTimeWindows.of(Time.minutes(5)));
//        SingleOutputStreamOperator<BatteryHealthResult> process = batteryHealthDataSingleOutputStreamOperator.keyBy(o -> "all").process(new BatteryHealthFunction());
        SingleOutputStreamOperator<String> apply = window.apply(new BatteryHealthWindowFunction());
        JdbcUtil.sinkToClickhouseUpsert(apply,
                "INSERT INTO vehicle_real.vehicle_dws_battery_Health(\n" +
                        "startTime, endTime, vehicleCount, sumTemp, avgTemp, ts)\n" +
                        "VALUES(?,?,?,?,?,?)");
    }
}
